
package com.jzo2o.foundations.handler;

import com.jzo2o.canal.listeners.AbstractCanalRabbitMqMsgListener;
import com.jzo2o.es.core.ElasticSearchTemplate;
import com.jzo2o.foundations.model.domain.ServeSync;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;

@Component
public class ServeCanalDataSyncHandler extends AbstractCanalRabbitMqMsgListener<ServeSync> {
    @Resource
    private ElasticSearchTemplate elasticSearchTemplate;

    //监听
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "canal-mq-jzo2o-foundations", arguments = {@Argument(name = "x-single-active-consumer", value = "true", type = "java.lang.Boolean")}),//消息队列
            exchange = @Exchange(name = "exchange.canal-jzo2o", type = ExchangeTypes.TOPIC),//交换机
            key = "canal-mq-jzo2o-foundations"
    ), concurrency = "1")//指定消费线程个数为1
    public void onMessage(Message message) throws Exception {
        System.out.println("监听消息。。。。"+message);
        //调用抽象类中的方法
        parseMsg(message);
    }

    //向es中保存数据。解析到binlog中的新增，更新消息执行此方法
    @Override
    public void batchSave(List<ServeSync> data) {
        //向es中保存索引
        Boolean serve_aggregation = elasticSearchTemplate.opsForDoc().batchInsert("serve_aggregation", data);
        //如果执行失败，要抛出异常，给mq回nack
        if (!serve_aggregation) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            throw new RuntimeException("同步数据失败");
        }
    }

    //将es中指定的文档进行删除
    @Override
    public void batchDelete(List<Long> ids) {
        //向es中保存索引
        Boolean serve_aggregation = elasticSearchTemplate.opsForDoc().insert("serve_aggregation", ids);
        //如果执行失败，要抛出异常，给mq回nack
        if (!serve_aggregation) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            throw new RuntimeException("同步数据失败");
        }
    }
}

